package com.elcom.roaming.common.util.match;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;

import com.elcom.roaming.common.config.ConfigModule;
import com.elcom.roaming.module.collection.match.UIMessage;
import com.elcom.roaming.module.collection.match.UIValue;
import com.elcom.roaming.module.synthesis.isup.ISUPMessage;
import com.elcom.roaming.module.synthesis.isup.ISUPValue;
import com.elcom.roaming.module.synthesis.match.MSRNMessage;
import com.elcom.roaming.module.synthesis.pr.PRMessage;
import com.elcom.roaming.module.synthesis.pr.PRValue;


public abstract class MatchingObjects implements Runnable {
	
	public static final int QUEUE_POLL_THREAD = 1;
	public static final int MAP_FULL = -1;
	public static final int NOT_FOUND_PUT = 1;
	public static final int FOUND_FORWARD = 2;
	
	private String name;
	private BlockingQueue<ObjectInput> queue;
	private ConcurrentHashMap<ObjectKey, ObjectValue> map;
	private String queueName;
	private String mapName;

	protected Logger logger;
	
	
	public MatchingObjects(Logger logger, String name) {
		this.logger = logger;
		this.name = name;
		this.queue = new LinkedBlockingQueue<ObjectInput>();
		this.map = new ConcurrentHashMap<ObjectKey, ObjectValue>();
		new Thread(new CleaningMap(logger, map), name+"_cleaning").start();
	}
		
	public abstract Matching getMatching();
	
	public String getName(){
		return name;
	}
	
	public BlockingQueue<ObjectInput> getQueue() {
		return queue;
	}

	public ConcurrentHashMap<ObjectKey, ObjectValue> getMap() {
		return map;
	}

	public String getQueueName() {
		return queueName;
	}

	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	public String getMapName() {
		return mapName;
	}

	public void setMapName(String mapName) {
		this.mapName = mapName;
	}

	public void put(ObjectInput obj){
		String mtype = null;
		if (obj instanceof ISUPMessage) mtype = "ISUP";
		else if (obj instanceof PRMessage) mtype = "PRN";
		else if (obj instanceof UIMessage) mtype = "UI";

		if (queue.size() >= ConfigModule.MATCHING_MAX_QUEUE_SIZE) {
			logger.info("REJECTED - " + mtype + " Matching Queue is full | " + obj.toString());
		} else{
			logger.info("Put into " + mtype + " matching queue: " + obj.toString());
			queue.offer(obj);
		}
	}
	
	public static int matchObject(ObjectInput obj, ConcurrentHashMap<ObjectKey, ObjectValue> matchMap) throws Exception{
		ObjectKey inputKey = obj.getKey();
		ObjectValue inputValue = obj.getValue();
		
		// put immediately : IAM, U0, P0
		if (obj instanceof UIMessage) {
			if (inputValue.getType() == UIValue.U0){
				matchMap.put(inputKey, inputValue);
				return NOT_FOUND_PUT;
			}
		} else if (obj instanceof ISUPMessage) {
			if (inputValue.getType() == ISUPValue.IAM){
				matchMap.put(inputKey, inputValue);
				return NOT_FOUND_PUT;
			}
		} else if (obj instanceof PRMessage) {
			if (inputValue.getType() == PRValue.P0){
				matchMap.put(inputKey, inputValue);
				return NOT_FOUND_PUT;
			}
		} 
		
		ObjectValue dropValue = matchMap.remove(inputKey);
		
		if (dropValue == null){
			if (matchMap.size() >= ConfigModule.MATCHING_MAX_MAP_SIZE){
				return MAP_FULL;
			} else {
				if (obj instanceof MSRNMessage) matchMap.put(inputKey, inputValue);
				return NOT_FOUND_PUT;
			}
		}
		else {
			obj.setDropValue(dropValue);
			return FOUND_FORWARD;
		}
	}
	
	@Override
	public void run() {
		logger.info("Started...");
		for (int i=0; i<QUEUE_POLL_THREAD; i++) {
			Matching matching = getMatching();
			matching.setQueue(queue);
			matching.setMap(map);
			matching.setLogger(logger);
			matching.setName(name+"_poll_"+i);
			new Thread(matching, matching.getName()).start();
		}
	}

}
